package com.yanggu.flink.datastream_api.offlinealarm

import cn.hutool.core.collection.CollUtil
import cn.hutool.core.date.DateUtil
import cn.hutool.json.JSONUtil
import org.apache.flink.api.common.state.{MapStateDescriptor, ValueStateDescriptor}
import org.apache.flink.streaming.api.functions.co.{BroadcastProcessFunction, KeyedCoProcessFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import java.util.Date
import scala.beans.BeanProperty

//如果一个设备有多个告警规则
//数据流和配置流进行connect, 判定设备是否离线
//三个测试数据
//{"deviceId":"deviceId","delete":false,"systemCode":"systemCode","delayTime":15000, "id": 1}
//{"deviceId":"deviceId","delete":false,"systemCode":"systemCode","delayTime":20000, "id": 2}
//{"cimCode":"cimCode","deviceId":"deviceId","systemCode":"systemCode","time":1643346340607,"value":0}
object OffAlarmDemo02 {

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    environment.setParallelism(1)

    //设备测点数据流
    val dataStream = environment
      .socketTextStream("localhost", 8000)
      .map(data => JSONUtil.toBean(data, classOf[CimPoint2]))

    //设备离线规则流
    val configDataStream = environment
      .socketTextStream("localhost", 9000)
      .map(data => JSONUtil.toBean(data, classOf[DeviceConfig2]))

    //key为二元组(systemCode, deviceId)表示一个设备, value为Set[Long]表示多个离线告警规则。也就是说一个设备可以有多个离线告警规则
    val mapStateDescriptor = new MapStateDescriptor[(String, String), java.util.Set[Long]]("mapStateDescriptor", classOf[(String, String)], classOf[java.util.Set[Long]])

    //广播流
    val broadCastStream = configDataStream.broadcast(mapStateDescriptor)

    dataStream
      //数据流connect广播流
      .connect(broadCastStream)
      //这里可以理解为给CimPoint数据打上一个ruleId的标签
      .process(new BroadcastProcessFunction[CimPoint2, DeviceConfig2, CimPoint2] {

        //处理主流数据
        override def processElement(value: CimPoint2, ctx: BroadcastProcessFunction[CimPoint2, DeviceConfig2, CimPoint2]#ReadOnlyContext, out: Collector[CimPoint2]): Unit = {
          val map = ctx.getBroadcastState(mapStateDescriptor)
          val key = (value.systemCode, value.deviceId)
          val offLineAlarmRuleIdSet = map.get(key)
          if (CollUtil.isEmpty(offLineAlarmRuleIdSet)) {
            return
          }
          offLineAlarmRuleIdSet.forEach(offLineAlarmRuleId => {
            //补充离线告警规则id
            value.offLineAlarmRuleId = offLineAlarmRuleId
            //向下游输出
            out.collect(value)
          })
        }

        //处理广播数据
        override def processBroadcastElement(value: DeviceConfig2, ctx: BroadcastProcessFunction[CimPoint2, DeviceConfig2, CimPoint2]#Context, out: Collector[CimPoint2]): Unit = {
          val map = ctx.getBroadcastState(mapStateDescriptor)
          val key = (value.systemCode, value.deviceId)
          var ids = map.get(key)
          //如果是删除数据告警规则, 从广播状态中删除
          if (value.delete && CollUtil.isNotEmpty(ids)) {
            ids.remove(value.id)
            return
          }
          if (CollUtil.isEmpty(ids)) {
            ids = new java.util.HashSet[Long]
          }
          ids.add(value.id)
          map.put(key, ids)
        }
      })
      //根据offLineAlarmRuleId进行分组
      .keyBy(_.offLineAlarmRuleId)
      //配置流也根据offLineAlarmRuleId进行分组, 这样测点数据和离线告警规则数据就放在一起了
      .connect(configDataStream.keyBy(_.id))
      //核心业务逻辑, 生产离线告警数据
      .process(new KeyedCoProcessFunction[Long, CimPoint2, DeviceConfig2, String] {

        /**
         * 最大离线时间
         */
        private lazy val maxOffLineTimeState = getRuntimeContext.getState(new ValueStateDescriptor[Long]("maxOffLineTimeState", classOf[Long]))

        /**
         * 最新数据的时间戳
         */
        private lazy val dataTimeState = getRuntimeContext.getState(new ValueStateDescriptor[Long]("dataTimeState", classOf[Long]))

        /**
         * 定时器执行时间戳
         */
        private lazy val triggerTimeStampState = getRuntimeContext.getState(new ValueStateDescriptor[Long]("triggerTimeStampState", classOf[Long]))

        /**
         * 是否离线的状态, true表示离线, false表示在线
         * 默认为true表示离线, 用于恢复之前的离线告警
         */
        private lazy val offLineState = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("offLineState", classOf[Boolean], true))

        //处理测点数据
        override def processElement1(value: CimPoint2,
                                     ctx: KeyedCoProcessFunction[Long, CimPoint2, DeviceConfig2, String]#Context,
                                     out: Collector[String]): Unit = {
          val maxOffLineTime = maxOffLineTimeState.value()
          val triggerTime = triggerTimeStampState.value()
          //判断是否有离线规则, 如果没有直接return
          if (maxOffLineTime == 0 && triggerTime == 0) {
            return
          }
          //如果数据的时间大于上次的时间就更新dataTimeState
          val time = value.time
          if (time > dataTimeState.value()) {
            dataTimeState.update(time)
            //如果设置之前离线了, 现在设备上传数据了发送数据表示设备在线了, 恢复之前的离线告警
            if (offLineState.value()) {
              offLineState.update(false)
              out.collect(s"离线告警规则id: ${ctx.getCurrentKey}, 重新上传数据时间: ${DateUtil.formatDateTime(new Date(time))}" +
                s", 恢复之前所有的离线告警, 当前时间: ${DateUtil.formatDateTime(new Date())}")
            }
            //删除之前的定时器, 重新注册定时器, 保存定时器执行时间
            ctx.timerService().deleteProcessingTimeTimer(triggerTime)
            val newTriggerTime = System.currentTimeMillis() + maxOffLineTimeState.value()
            ctx.timerService().registerProcessingTimeTimer(newTriggerTime)
            triggerTimeStampState.update(newTriggerTime)
          }
        }

        //处理离线告警规则
        override def processElement2(value: DeviceConfig2,
                                     ctx: KeyedCoProcessFunction[Long, CimPoint2, DeviceConfig2, String]#Context,
                                     out: Collector[String]): Unit = {
          val triggerTime = triggerTimeStampState.value()
          val timerService = ctx.timerService()

          //如果是删除离线告警规则, 清空所有状态和销毁定时器
          if (value.delete) {
            maxOffLineTimeState.clear()
            triggerTimeStampState.clear()
            dataTimeState.clear()
            offLineState.clear()
            timerService.deleteProcessingTimeTimer(triggerTime)
            println(s"删除离线告警规则: ${ctx.getCurrentKey}, 清空所有状态和定时器")
            return
          }

          val lastMaxOffLineTime = maxOffLineTimeState.value()
          val currentDelayTime = value.delayTime

          //如果是第一次设置时间, 注册定时器, 更新定时器的执行时间
          if (lastMaxOffLineTime == 0 && triggerTime == 0) {
            val triggerTimeStamp = System.currentTimeMillis() + currentDelayTime
            timerService.registerProcessingTimeTimer(triggerTimeStamp)
            triggerTimeStampState.update(triggerTimeStamp)
            //如果设置时间发生了变化且不是第一次更新时间
          } else if (lastMaxOffLineTime != 0 && triggerTime != 0 && lastMaxOffLineTime != currentDelayTime) {
            //删除原有的定时器
            timerService.deleteProcessingTimeTimer(triggerTime)
            val newTriggerTime = triggerTime - lastMaxOffLineTime + currentDelayTime
            //注册新的定时器
            timerService.registerProcessingTimeTimer(newTriggerTime)
            //更新定时器执行的时间戳
            triggerTimeStampState.update(newTriggerTime)
          }

          //直接更新状态即可
          maxOffLineTimeState.update(currentDelayTime)
        }

        override def onTimer(timestamp: Long,
                             ctx: KeyedCoProcessFunction[Long, CimPoint2, DeviceConfig2, String]#OnTimerContext,
                             out: Collector[String]): Unit = {
          val dataTime = dataTimeState.value()
          val delay = maxOffLineTimeState.value()
          if (dataTime + delay < timestamp) {
            out.collect(s"离线告警规则id: ${ctx.getCurrentKey}, 上一次上传数据的时间: ${DateUtil.formatDateTime(new Date(dataTime))}, " +
              s"离线了${(timestamp - dataTime) / 1000}秒, 离线时间: ${DateUtil.formatDateTime(new Date(timestamp))}")
            //更新状态表示离线了
            offLineState.update(true)
          }

          //继续注册定时器, 检测是否离线
          val triggerTimeStamp = timestamp + delay
          ctx.timerService().registerProcessingTimeTimer(triggerTimeStamp)
          triggerTimeStampState.update(triggerTimeStamp)
        }
      })
      .print("OffLineAlarm02>>>>>")

    environment.execute("OffAlarmDemo02 Job")

  }

}

case class CimPoint2(@BeanProperty var systemCode: String,
                     @BeanProperty var deviceId: String,
                     @BeanProperty var cimCode: String,
                     @BeanProperty var time: Long,
                     @BeanProperty var value: Double,
                     @BeanProperty var offLineAlarmRuleId: Long)

case class DeviceConfig2(@BeanProperty var id: Long,
                         @BeanProperty var systemCode: String,
                         @BeanProperty var deviceId: String,
                         @BeanProperty var delayTime: Long,
                         @BeanProperty var delete: Boolean)